package com.atguigu.flink.chapter08;




import com.atguigu.flink.bean.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Flink01_Project_Product_PV {

       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
               env.setParallelism(1);

               env.readTextFile("input/UserBehavior.csv")
                       .map(line->{
                           String[] datas = line.split(",");
                           return new UserBehavior(
                                   Long.valueOf(datas[0]),
                                   Long.valueOf(datas[1]),
                                   Integer.valueOf(datas[2]),
                                   datas[3],
                                   Long.parseLong(datas[4])*1000
                           );
                       })
                       // 添加水印
                       .assignTimestampsAndWatermarks(
                               WatermarkStrategy
                                       .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                       //添加时间戳
                               .withTimestampAssigner((ub,ts)->ub.getTimestamp())
                       )
                       // 过滤行为，只要 pv 行为
                       .filter(ub ->"pv".equals(ub.getBehavior()))
                       .keyBy(UserBehavior::getBehavior)
                       .window(TumblingEventTimeWindows.of(Time.minutes(30)))
                       .aggregate(new AggregateFunction<UserBehavior, Long, Long>() {
                                      @Override
                                      public Long createAccumulator() {
                                          return 0L;
                                      }

                                      @Override
                                      public Long add(UserBehavior value, Long acc) {
                                          return acc + 1;
                                      }

                                      @Override
                                      public Long getResult(Long acc) {
                                          return acc;
                                      }

                                      @Override
                                      public Long merge(Long a, Long b) {
                                          return null;
                                      }
                                  }, new ProcessWindowFunction<Long, String, String, TimeWindow>() {
                                      @Override
                                      public void process(String key,
                                                          Context ctx
                                                        , Iterable<Long> elements,
                                                          Collector<String> out) throws Exception {
                                          out.collect(ctx.window() + " " + elements.iterator().next());
                                      }
                                  }
                       )
                       .print();






               try {
                   env.execute();
               } catch (Exception e) {
                   e.printStackTrace();
               }


           }


}
